package snippets.cse524.activeDates;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;

import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class ThresholdedIntervalsMR extends Configured
		implements Tool {

	public static class Map extends MapReduceBase implements
			Mapper<LongWritable, Text, Text, Text> {
		private static final IntWritable ONE = new IntWritable(1);
		public int[] GAPS = { 1, 5, 10, 15, 20 };

		@Override
		public void map(LongWritable key, Text line,
				OutputCollector<Text, Text> collector, Reporter arg3)
				throws IOException {

			String content = line.toString();
			
			String [] parts = StringUtils.split(content, '\t');
			
			String pageId = parts[0];
			double mentions = Double.parseDouble(parts[1]);
			
			java.util.Map<Integer, Double> yearCounts = new HashMap<Integer, Double>();
			
			for(String part : ArrayUtils.subarray(parts, 2, parts.length)) {
				String [] ycParts = StringUtils.split(part, ':');
				
				yearCounts.put(Integer.parseInt(ycParts[0]), Double.parseDouble(ycParts[1]));
			}
			
			List<Double> thresholds = new ArrayList<Double>();

			thresholds.add(0.6);
			thresholds.add(0.7);
			thresholds.add(0.8);
			thresholds.add(0.9);
			thresholds.add(0.95);
			thresholds.add(1.0d);
			
			java.util.Map<Double, Pair<Integer, Integer>> res = snippets.utils.Utils.getMinimumIntervalsForThresholds(thresholds, yearCounts);
			
			List<String> opParts = new ArrayList<String>();
			
			for(Double _threshold : thresholds) {
				if(!res.containsKey(_threshold)) continue;
				
				opParts.add(Double.toString(_threshold));
				
				opParts.add(Integer.toString(res.get(_threshold).getLeft()));
				opParts.add(Integer.toString(res.get(_threshold).getRight()));
			}
			
			collector.collect(new Text(pageId), new Text(StringUtils.join(opParts, '\t')));
		}
	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
		ToolRunner.run(new ThresholdedIntervalsMR(), args);

	}

	@Override
	public int run(String[] args) throws Exception {
		String inputPath = args[0];
		String outputPath = args[1];

		JobConf conf = new JobConf(getConf(),
				ThresholdedIntervalsMR.class);
		conf.setJobName(String.format("ThresholdedIntervalsMR [%s: %s]",
				inputPath, outputPath));

		conf.setNumMapTasks(12);
		conf.setNumReduceTasks(12);

		FileInputFormat.setInputPaths(conf, new Path(inputPath));
		FileOutputFormat.setOutputPath(conf, new Path(outputPath));

		conf.setInputFormat(TextInputFormat.class);
		conf.setOutputFormat(TextOutputFormat.class);

		conf.setMapperClass(Map.class);

		// Set the outputs for the Map
		// PairOfWritables<ArrayListWritable<PairOfStrings>,
		// ArrayListWritable<PairOfWritables<Text,
		// ArrayListWritable<PairOfWritables<DoubleWritable,
		// PairOfWritables<IntWritable, PairOfInts>>>>>> mapOpDummy = new
		// PairOfWritables<ArrayListWritable<PairOfStrings>,
		// ArrayListWritable<PairOfWritables<Text,
		// ArrayListWritable<PairOfWritables<DoubleWritable,
		// PairOfWritables<IntWritable, PairOfInts>>>>>>();
		conf.setMapOutputKeyClass(Text.class);
		conf.setMapOutputValueClass(Text.class);

		JobClient.runJob(conf);

		return 0;
	}
}
